/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.messaging.rocket.consumer;

import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.ipig.commons.codec.CodecService;
import org.ipig.commons.codec.kryo.GenericKryoCodec;
import org.ipig.commons.conf.rocket.consumer.RocketConsumerConf;
import org.ipig.commons.service.ReporterService;
import org.ipig.commons.service.monitor.DoneMsgsEvent;
import org.ipig.constants.ResultStatus;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.Resource;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.LongAdder;

/**
 * 抽象推消息消费
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @version $$Id$$
 * @since 1.0
 */
@Slf4j
public abstract class AbstractRocketPushConsumer implements RocketConsumerService {
    //编码
    @Setter
    @Getter
    @Resource(type = GenericKryoCodec.class)
    protected CodecService codecService;
    @Setter
    @Getter
    @Autowired(required = false)
    protected RocketConsumerConf conf;
    //汇报器服务
    @Setter
    @Getter
    @Autowired(required = false)
    private ReporterService<DoneMsgsEvent, Void> reporterService;
    //flag
    private volatile boolean flag = false;
    //处理消息数量
    private LongAdder msgsTotal = new LongAdder();
    //开始时间 时间戳
    private volatile long beginTimeMills;

    @Override
    public Void execute() {
        init();
        return null;
    }


    /**
     * init
     */
    public void init() {
        if (flag) {
            return;
        }
        synchronized (this) {
            if (flag) {
                return;
            }
            if (conf == null) {
                throw new RuntimeException("configure the KafkaConsumerConf instance");
            }
            try {
                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
                consumer.setNamesrvAddr(conf.getNamesrvAddr());
                consumer.setInstanceName(conf.getInstanceName());
                consumer.setConsumerGroup(conf.getGroup());
                consumer.subscribe(conf.getSubscribeTopic(), conf.getSubscribeExpression());
                /**
                 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费<br>
                 * 如果非第一次启动，那么按照上次消费的位置继续消费
                 */
                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                /* 异步接收，异步传递的消息 */
                consumer.registerMessageListener(new MessageListenerConcurrently() {
                    @Override
                    public ConsumeConcurrentlyStatus consumeMessage(
                            List<MessageExt> msgList, ConsumeConcurrentlyContext context) {
                        if (msgList != null && !msgList.isEmpty()) {
                            if (beginTimeMills <= 0) {
                                beginTimeMills = System.currentTimeMillis();
                            }
                            try {
                                if (doProcess(msgList) == ResultStatus.SUCCESS) {
                                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
                                }
                            } catch (Exception ex) {
                                ex.printStackTrace();
                                log.error(ExceptionUtils.getFullStackTrace(ex));
                                System.err.println(msgList);
                            } finally {
                                // 记录处理消息的总量，不管处理成功与失败
                                msgsTotal.add(msgList.size());
                            }
                            return ConsumeConcurrentlyStatus.RECONSUME_LATER;// 重试
                        } else {
                            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;// 成功
                        }
                    }
                });
                consumer.start();
                Thread.sleep(5000);
                flag = true;
                ScheduledExecutorService scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
                scheduledExecutor.scheduleAtFixedRate(() -> {
                    if (reporterService != null) {
                        DoneMsgsEvent doneMsgs = new DoneMsgsEvent();
                        doneMsgs.setDoneMsgsTotal(msgsTotal.longValue());
                        doneMsgs.setTopic(conf.getSubscribeTopic());
                        doneMsgs.setBeginTimeMills(beginTimeMills);
                        doneMsgs.setEndTimeMills(System.currentTimeMillis());
                        reporterService.report(doneMsgs);
                    } else {
                        if (beginTimeMills > 0
                                && msgsTotal.longValue() > 0) {
                            log.info("总用时:{}.消息量:{}", (System.currentTimeMillis() - beginTimeMills) / 1000, msgsTotal.longValue());
                        }
                    }
                }, 10, 30, TimeUnit.SECONDS);
            } catch (MQClientException e) {
                log.error("MQClientException业务处理出错", e);
            } catch (Exception ex) {
                log.error("业务处理出错", ex);
            }
        }
    }

    /**
     * <b>业务处理</b>
     * <li>如果返回FAILURE或抛出异常， 就回滚事务【<b>特别提醒：业务处理要做幂等</b>】</li>
     * <li>如果返回SUCCESS ，就提交事务</li>
     *
     * @param msgList List<MessageExt>
     * @return ResultStatus
     */
    public abstract ResultStatus doProcess(List<MessageExt> msgList);
}

